package transform;

import beans.SenSorReading;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class reduceTest01 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //从文件中读取数据
        DataStreamSource<String> inputStream = env.readTextFile("src/main/resources/sensor.txt");

        SingleOutputStreamOperator<SenSorReading> dataStream = inputStream.map(line -> {
            String[] fields = line.split(",");
            return new SenSorReading(fields[0], new Long(fields[1]), new Double(fields[2]));
        });

        KeyedStream<SenSorReading, String> keyByStream = dataStream.keyBy(data -> data.getId());

        //reduce 取得是最大的温度值，和最新的时间戳
//        SingleOutputStreamOperator<SenSorReading> reduceStream = keyByStream.reduce(new ReduceFunction<SenSorReading>() {
//            /**
//             * 第一个是之前处理的结果数据
//             * 第二个是等等待处理的数据
//             * @param senSorReading
//             * @param t1
//             * @return
//             * @throws Exception
//             */
//            @Override
//            public SenSorReading reduce(SenSorReading senSorReading, SenSorReading t1) throws Exception {
//                return new SenSorReading(senSorReading.getId(), t1.getTimeStamp(), Math.max(senSorReading.getTemperature(), t1.getTemperature()));
//            }
//        });
        SingleOutputStreamOperator<SenSorReading> reduceStream =keyByStream.reduce((curr, newData) -> new SenSorReading(curr.getId(), newData.getTimeStamp(), Math.max(curr.getTemperature(), newData.getTemperature())));
        reduceStream.print();
        env.execute();
    }
}
